package cn.gwm.flink.streaming.dwm.charge.function;

import cn.gwm.flink.streaming.constant.BaseFields;
import cn.gwm.flink.streaming.constant.VehicleType;
import cn.gwm.flink.streaming.dwm.charge.bean.ChargeArgs;
import cn.gwm.flink.streaming.dwm.charge.bean.ChargeFavourable;
import cn.gwm.flink.streaming.dwm.charge.bussiness.ChargeHandle;
import cn.gwm.flink.streaming.dwm.charge.model.ChargeConstant;
import cn.gwm.flink.streaming.dwm.charge.model.ChargeStateEnum;
import cn.gwm.flink.streaming.dwm.charge.support.ChargeState;
import cn.gwm.utils.ConfigLoader;
import cn.hutool.json.JSONObject;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.*;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector;

import java.util.List;

/**
 * @ClassName ChargeFunction
 * @Description
 * @Author LiangGuang
 * @Date 2023/05/10 11:05
 */
public class ChargeFunction extends RichFlatMapFunction<JSONObject, JSONObject> {


    private ChargeConstant constant;

    private Boolean tripRedis = Boolean.FALSE;

    public ChargeFunction(ChargeConstant constant) {
        this.constant = constant;
    }

    public ChargeFunction(ChargeConstant constant, Boolean tripRedis) {
        this.constant = constant;
        this.tripRedis = tripRedis;
    }

    /**
     * 数据缓存
     */
    private ListState<JSONObject> cacheListState;

    /**
     * 当前稳态 状态
     */
    private ValueState<ChargeStateEnum> steadyState;

    /**
     * 电流中断时间
     */
    private ValueState<Long> chargeBreakState;
    /**
     * 有效充电时长
     */
    private ValueState<Long> effectiveChargeTime;

    /**
     * 波动状态的记录，list中，存入波动上一条数据和波动的当前数据
     */
    private MapState<String, List> mapState;
    /**
     * 上一条数据
     */
    private ValueState<JSONObject> preDataState;

    /**
     * 一级汇总数据
     */
    private ValueState<JSONObject> summarState;
    /**
     * 二级汇总数据
     */
    private ValueState<JSONObject> bandSummarState;
    /**
     * 插枪时间 --》 确认过稳态
     */
    private ValueState<Long> ConnectTimeState;
    /**
     * 首次插枪记录 --> 非稳态
     */
    private ValueState<Long> firstConnectTimeState;
    /**
     * 充电过程中优惠信息的存储
     */
    private MapState<Long, ChargeFavourable> favourableMapState;
    /**
     * 记录充电模式首条 电量值
     */
    private MapState<String, JSONObject> chargeModelFirstState;


    @Override
    public void open(Configuration parameters) throws Exception {
        cacheListState = getRuntimeContext().getListState(new ListStateDescriptor<>("cacheListState", JSONObject.class));
        chargeBreakState = getRuntimeContext().getState(new ValueStateDescriptor<>("chargeBreakState", Long.class));
        steadyState = getRuntimeContext().getState(new ValueStateDescriptor<>("steadyState", ChargeStateEnum.class));

        effectiveChargeTime = getRuntimeContext().getState(new ValueStateDescriptor<>("effectiveChargeTime", Long.class));
        ConnectTimeState = getRuntimeContext().getState(new ValueStateDescriptor<>("ConnectTimeState", Long.class));
        firstConnectTimeState = getRuntimeContext().getState(new ValueStateDescriptor<>("firstConnectTimeState", Long.class));

        preDataState = getRuntimeContext().getState(new ValueStateDescriptor<>("preDataState", JSONObject.class));
        summarState = getRuntimeContext().getState(new ValueStateDescriptor<>("summarState", JSONObject.class));
        bandSummarState = getRuntimeContext().getState(new ValueStateDescriptor<>("bandSummarState", JSONObject.class));
        mapState = getRuntimeContext().getMapState(new MapStateDescriptor<>("mapState", String.class, List.class));
        favourableMapState = getRuntimeContext().getMapState(new MapStateDescriptor<>("favourableMapState", Long.class, ChargeFavourable.class));
        chargeModelFirstState = getRuntimeContext().getMapState(new MapStateDescriptor<>("chargeModelFirstState", String.class, JSONObject.class));

    }

    @Override
    public void flatMap(JSONObject nowData, Collector<JSONObject> collector) throws Exception {
        ChargeStateEnum chargeState = ChargeState.realState(constant, nowData, summarState);
        // 单独处理插枪
        // 如果 当前不是物理断开状态且 插枪时间不存在，则记录
        ConnectTimeUpdate(ConnectTimeState, chargeState, nowData, firstConnectTimeState);
        // 构造充电参数
        ChargeArgs args = new ChargeArgs()
                .setRealNowData(nowData)
                .setOut(collector)
                .setBandSummarState(bandSummarState)
                .setChargeBreakState(chargeBreakState)
                .setConnectTimeState(ConnectTimeState)
                .setEffectiveChargeTime(effectiveChargeTime)
                .setListState(cacheListState)
                .setSummarState(summarState)
                .setNowState(chargeState)
                .setFavourableMapState(favourableMapState)
                .setChargeModelFirstState(chargeModelFirstState)
                .setConstant(constant)
                .setTripRedis(tripRedis);
        if (nowData.get(VehicleType.VEHICLE_TYPE).equals(VehicleType.ES13)) {
            args.setPhoenixTableName(ConfigLoader.get("hbase.ods.table.name.gps.es13"));
        } else if (nowData.get(VehicleType.VEHICLE_TYPE).equals(VehicleType.V71_PHEV)) {
            args.setPhoenixTableName(ConfigLoader.get("hbase.ods.table.name.gps.v71"));
        } else if (nowData.get(VehicleType.VEHICLE_TYPE).equals(VehicleType.P03_PHEV)) {
            String p03Table = ConfigLoader.get("hbase.ods.table.name.gps.p03");
            String dataBase = p03Table.substring(0, p03Table.indexOf(":"));
            String tableName = p03Table.substring(p03Table.indexOf(":") + 1, p03Table.length());
            args.setDataBase(dataBase);
            args.setPhoenixTableName(tableName);
        }
        ChargeHandle.charge(preDataState, steadyState, mapState, chargeState, args);
        preDataState.update(nowData);

    }


    public static void ConnectTimeUpdate(ValueState<Long> ConnectTimeState, ChargeStateEnum chargeState,
                                         JSONObject nowData, ValueState<Long> firstConnectTimeState) throws Exception {
        // 插枪的稳态在单字段已经校验过了，此处只需要判断是否为拔枪状态即可
        if (ConnectTimeState.value() == null && !ChargeStateEnum.END.equals(chargeState)) {
            ConnectTimeState.update(nowData.getLong(BaseFields.tid));
        }
    }

}
